﻿using System.Collections.Concurrent;
using System.Text;

using Etcdserverpb;

using Google.Protobuf;

using Grpc.Core;
using Grpc.Net.Client;
using Grpc.Net.Client.Balancer;
using Grpc.Net.Client.Configuration;

using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;

using static Etcdserverpb.Watch;

namespace Jack.EtcdConfiguration
{
    public class EtcdConfigurationProvider : ConfigurationProvider, IConfigurationSource, IDisposable
    {
        public static CancellationToken CancellationToken { get; set; } = CancellationToken.None;

        private readonly EtcdOptions _etcdClientOptions;
        private readonly bool _reloadOnChange;
        private readonly Action<IConfigurationRoot> _actionOnChange;
        private readonly SocketsHttpHandler _defaultHttpHandler = new SocketsHttpHandler
        {
            PooledConnectionIdleTimeout = TimeSpan.FromMinutes(5),
            KeepAlivePingDelay = TimeSpan.FromSeconds(30),
            KeepAlivePingTimeout = TimeSpan.FromSeconds(30),
            KeepAlivePingPolicy = HttpKeepAlivePingPolicy.Always,
            EnableMultipleHttp2Connections = true,
        };
        private readonly MethodConfig _defaultGrpcMethodConfig = new()
        {
            Names = { MethodName.Default },
            RetryPolicy = new RetryPolicy
            {
                MaxAttempts = 5,
                InitialBackoff = TimeSpan.FromSeconds(1),
                MaxBackoff = TimeSpan.FromSeconds(5),
                BackoffMultiplier = 1.5,
                RetryableStatusCodes = { StatusCode.Unavailable }
            }
        };
        private readonly RetryThrottlingPolicy _defaultRetryThrottlingPolicy = new()
        {
            MaxTokens = 10,
            TokenRatio = 0.1
        };
        private readonly GrpcChannel _grpcChannel;
        private Metadata _headers;
        private Etcdserverpb.KV.KVClient _kVClient = null;
        private Etcdserverpb.Watch.WatchClient _watchClient = null;
        private Etcdserverpb.Auth.AuthClient _authClient = null;

        public EtcdConfigurationProvider(EtcdOptions etcdClientOptions, bool reloadOnChange, Action<IConfigurationRoot> actionOnChange)
        {
            if (string.IsNullOrEmpty(etcdClientOptions.ConnectionString))
            {
                throw new ArgumentNullException($"{nameof(etcdClientOptions.ConnectionString)} 不能为空");
            }

            if (etcdClientOptions.PrefixKeys == null || !etcdClientOptions.PrefixKeys.Any())
            {
                throw new ArgumentNullException($"{nameof(etcdClientOptions.PrefixKeys)} 不能为空");
            }

            _etcdClientOptions = etcdClientOptions;
            _reloadOnChange = reloadOnChange;
            _actionOnChange = actionOnChange;

            #region 构建grpc通道
            GrpcChannelOptions grpcChannelOptions = new GrpcChannelOptions
            {
                ServiceConfig = new ServiceConfig
                {
                    MethodConfigs = { _defaultGrpcMethodConfig },
                    RetryThrottling = _defaultRetryThrottlingPolicy,
                    LoadBalancingConfigs = { new RoundRobinConfig() }
                },
                HttpHandler = _defaultHttpHandler,
                DisposeHttpClient = true,
                ThrowOperationCanceledOnCancellation = true,
                Credentials = ChannelCredentials.Insecure
            };

            string[] hosts = etcdClientOptions.ConnectionString.Split(',');
            List<Uri> nodes = new();
            foreach (string host in hosts)
            {
                string processedHost = host.Trim();
                nodes.Add(new Uri(processedHost));
            }

            StaticResolverFactory factory =
                new(addr => nodes.Select(i => new BalancerAddress(i.Host, i.Port)).ToArray());
            ServiceCollection services = new();
            services.AddSingleton<ResolverFactory>(factory);
            grpcChannelOptions.ServiceProvider = services.BuildServiceProvider();
            _grpcChannel = GrpcChannel.ForAddress($"static://{Guid.NewGuid():n}", grpcChannelOptions);
            #endregion

            #region 如果配置了账号密码，获取授权用的token
            if (!string.IsNullOrEmpty(_etcdClientOptions.Username) && !string.IsNullOrEmpty(_etcdClientOptions.Password))
            {
                _authClient = new Etcdserverpb.Auth.AuthClient(_grpcChannel);
                var authenticateResponse = _authClient.Authenticate(new Etcdserverpb.AuthenticateRequest
                {
                    Name = _etcdClientOptions.Username,
                    Password = _etcdClientOptions.Password,
                });

                _headers = new Metadata
                {
                    { "Authorization",  authenticateResponse.Token }
                };
            }
            #endregion

            if (reloadOnChange || actionOnChange != null)
            {
                Watch();
            }
        }

        public override void Load()
        {
            _kVClient ??= new Etcdserverpb.KV.KVClient(_grpcChannel);

            var configMap = new ConcurrentDictionary<string, string>(StringComparer.OrdinalIgnoreCase);
            string env = _etcdClientOptions.Env.TrimEnd('/');
            foreach (var prefixKey in _etcdClientOptions.PrefixKeys)
            {
                var fullPrefixKey = $"{env}/{prefixKey.Trim('/')}/";
                string rangeEnd = EtcdHelper.GetRangeEnd(fullPrefixKey);

                var rangeResponse = _kVClient.Range(new Etcdserverpb.RangeRequest
                {
                    Key = ByteString.CopyFromUtf8(fullPrefixKey),
                    RangeEnd = ByteString.CopyFromUtf8(rangeEnd),
                }, _headers);

                var kvs = rangeResponse.Kvs;
                foreach (var kv in kvs)
                {
                    var key = kv.Key.ToStringUtf8();
                    var val = kv.Value.ToStringUtf8();

                    key = key.Substring(fullPrefixKey.Length);
                    key = key.Replace("/", ":");
                    if (EtcdHelper.IsJsonString(val))
                    {
                        using MemoryStream ms = new MemoryStream(Encoding.UTF8.GetBytes(val));
                        var configuration = new ConfigurationBuilder()
                        .AddJsonStream(ms)
                        .Build();

                        foreach (var kvp in configuration.AsEnumerable())
                        {
                            configMap[$"{key}:{kvp.Key}"] = kvp.Value;
                        }
                    }
                    else
                    {
                        configMap[key] = val;
                    }
                }
            }

            Data.Clear();
            Data = configMap;
        }

        private void Reload()
        {
            Load();

            if (_actionOnChange != null)
            {
                var builder = new ConfigurationBuilder().AddInMemoryCollection(Data).Build();
                _actionOnChange(builder);
            }
        }

        public void FireChange()
        {
            Reload();
            OnReload();
        }

        public void Watch()
        {
            _watchClient ??= new Etcdserverpb.Watch.WatchClient(_grpcChannel);

            string env = _etcdClientOptions.Env.TrimEnd('/');
            foreach (var prefixKey in _etcdClientOptions.PrefixKeys)
            {
                string _prefixKey = prefixKey.Trim('/');

                Task.Run(async () =>
                {
                    var fullPrefixKey = $"{env}/{_prefixKey}/";
                    string rangeEnd = EtcdHelper.GetRangeEnd(fullPrefixKey);
                    var createRequest = new WatchCreateRequest
                    {
                        Key = ByteString.CopyFromUtf8(fullPrefixKey),
                        RangeEnd = ByteString.CopyFromUtf8(rangeEnd),
                    };

                    do
                    {
                        try
                        {
                            using var call = _watchClient.Watch(_headers);
                            await call.RequestStream.WriteAsync(new WatchRequest { CreateRequest = createRequest });
                            await call.RequestStream.CompleteAsync();

                            while (await call.ResponseStream.MoveNext())
                            {
                                if (CancellationToken.IsCancellationRequested) break;

                                var response = call.ResponseStream.Current;

                                if (response.Created)
                                {
                                    if (response.WatchId == -1) // 订阅失败，获取新的token
                                    {
                                        #region 如果配置了账号密码，获取授权用的token
                                        if (!string.IsNullOrEmpty(_etcdClientOptions.Username) && !string.IsNullOrEmpty(_etcdClientOptions.Password))
                                        {
                                            _authClient = new Etcdserverpb.Auth.AuthClient(_grpcChannel);
                                            var authenticateResponse = _authClient.Authenticate(new Etcdserverpb.AuthenticateRequest
                                            {
                                                Name = _etcdClientOptions.Username,
                                                Password = _etcdClientOptions.Password,
                                            });

                                            _headers = new Metadata
                                            {
                                                { "Authorization",  authenticateResponse.Token }
                                            };
#if DEBUG
                                            Console.WriteLine($"观察者创建失败（{DateTime.Now}），标识为: {response.WatchId}，监听的前缀key为：{fullPrefixKey}，重新获取token");
#endif
                                            break;
                                        }
                                        #endregion
                                    }
#if DEBUG
                                    Console.WriteLine($"观察者创建成功（{DateTime.Now}），标识为: {response.WatchId}，监听的前缀key为：{fullPrefixKey}");
#endif
                                }
                                else if (response.Canceled)
                                {
#if DEBUG
                                    Console.WriteLine($"观察者已被取消（{DateTime.Now}），标识为: {response.WatchId}，监听的前缀key为：{fullPrefixKey}");
#endif
                                    break;
                                }
                                else
                                {
#if DEBUG
                                    Console.WriteLine($"观察者收到修改通知（{DateTime.Now}），标识为: {response.WatchId}，监听的前缀key为：{fullPrefixKey}");
#endif
                                    FireChange();
                                }
                            }
                        }
                        catch (OperationCanceledException) { }
                        catch (Exception ex)
                        {
                            Console.WriteLine(ex);
                        }
                    } while (CancellationToken.IsCancellationRequested == false);
                }).ConfigureAwait(false);
            }

#if DEBUG
            Console.WriteLine("监听初始化完毕");
#endif
        }

        public IConfigurationProvider Build(IConfigurationBuilder builder) => this;

        #region IDisposable
        private bool _disposed;
        protected virtual void Dispose(bool disposing)
        {
            if (!_disposed)
            {
                if (disposing)
                {
                    _grpcChannel?.Dispose();
                }

                _disposed = true;
            }
        }
        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }
        #endregion
    }
}
